package cn.ctodb.im.logic.rpc;

import cn.ctodb.im.api.Session;
import cn.ctodb.im.dto.AbstractDTO;
import cn.ctodb.im.dto.MessageRespDTO;
import cn.ctodb.im.dto.MessageTextReqDTO;
import cn.ctodb.im.kafka.KafkaUtils;
import cn.ctodb.im.logic.peer.PeerManager;
import cn.ctodb.im.rpc.MessageRpc;
import com.alibaba.dubbo.config.annotation.Service;
import com.fasterxml.jackson.core.JsonProcessingException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.util.List;

/**
 * All rights Reserved, Designed By www.ctodb.cn
 *
 * @version V1.0
 * @author: lichaohn@163.com
 * @date: 2018-11-12 14:31
 * @Copyright: 2018 www.ctodb.cn All rights reserved.
 */
@Service
public class MessageRpcImpl implements MessageRpc {
    private Logger logger = LoggerFactory.getLogger(getClass());

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    @Autowired
    private PeerManager peerManager;

    @Override
    public MessageRespDTO sendText(MessageTextReqDTO textReq) {
        MessageRespDTO messageRespDTO = new MessageRespDTO();
        try {
            logger.debug("处理消息:{}", textReq);
            List<Session> sessions = peerManager.get(textReq.getTo());
            for (Session session : sessions) {
                try {
                    ListenableFuture listenableFuture = kafkaTemplate.send(KafkaUtils.toProducerRecord(session.getGateAddr(), session.getChannelid(), textReq));
                    listenableFuture.addCallback(new ListenableFutureCallback() {
                        @Override
                        public void onFailure(Throwable throwable) {

                        }

                        @Override
                        public void onSuccess(Object o) {

                        }
                    });
                    logger.debug("转发消息至:{}", session.getGateAddr());
                } catch (JsonProcessingException e) {
                    logger.error(e.getMessage(), e);
                }
            }
            messageRespDTO.setTs(System.currentTimeMillis());
            messageRespDTO.setId(System.currentTimeMillis());
            messageRespDTO.setCc(textReq.getCc());
        } catch (Exception e) {
            messageRespDTO.setStatus(AbstractDTO.Status.ERROR);
            messageRespDTO.setErrMsg(e.getMessage());
            logger.error(e.getMessage(), e);
        }
        return messageRespDTO;
    }
}
